/**
 * 
 */
package com.thomsonreuters.ce.base.io.queue;

/**
 * @author lei.yang
 *
 */
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.io.Serializable;

import com.thomsonreuters.ce.base.io.AdvancedBIS;
import com.thomsonreuters.ce.base.io.ObjectReadStream;
import com.thomsonreuters.ce.base.io.ObjectWriteStream;

public class DiskQueue {

	private int Size;

	private String tempDir;

	private String tempFileName=null;
	
	private long CacheReadPosition = 0L;
	
	private int Pointer=1;
		
	
	public DiskQueue(String TempDir)
	{
		this.tempDir = TempDir;
		this.Size=0;
		this.CacheReadPosition=0L;
		
		try {
			
			File tempFile = File.createTempFile("DiskQueue", null, new File(this.tempDir));	
			this.tempFileName=tempFile.getName();
			
		} catch (Exception e) {
			throw new MQException("Initialize DiskQueue failed: ",e);
			
		}				
	}

	public boolean isEmpty()
	{
		if (this.Size==0)
		{
			return true;
		}
				
		return false;
	}

	public int GetSize()
	{
		return this.Size;
	}

	public void Delete() throws IOException
	{
		File TempFile=new File(this.tempDir,this.tempFileName);
		TempFile.delete();
	}

	public Serializable RemoveFirst() throws IOException,ClassNotFoundException 
	{
		File TempFile=new File(this.tempDir,this.tempFileName);
		FileInputStream Fin=new FileInputStream(TempFile);
		Fin.getChannel().position(this.CacheReadPosition);
		AdvancedBIS ABIS=new AdvancedBIS(Fin);
		ObjectReadStream oI=new ObjectReadStream(ABIS);

		Serializable FirstObj=(Serializable)oI.readObject();
		
		this.CacheReadPosition=Fin.getChannel().position()-ABIS.getCount()+ABIS.getPos();

		Fin.close();
		
		this.Size--;
		
		if (this.Size==0)
		{
			TempFile.delete();
		}
		
		return FirstObj;
	}
	
	public Serializable GetNext() throws IOException,ClassNotFoundException
	{
		File TempFile=new File(this.tempDir,this.tempFileName);
		FileInputStream Fin=new FileInputStream(TempFile);
		Fin.getChannel().position(this.CacheReadPosition);
		AdvancedBIS ABIS=new AdvancedBIS(Fin);
		ObjectReadStream oI=new ObjectReadStream(ABIS);

		Serializable NextObj=(Serializable)oI.readObject();
		
		this.CacheReadPosition=Fin.getChannel().position()-ABIS.getCount()+ABIS.getPos();
		this.Pointer++;
		
		Fin.close();
		
		return NextObj;
	}
	
	public void ResetPointer()
	{
		this.CacheReadPosition=0L;		
		this.Pointer=1;
	}
	
	public boolean ReachEnd()
	{
		if (this.Pointer>this.Size)
		{
			return true;
		}
		else
		{
			return false;
		}
	}
	
	

	public void AddtoEnd(Serializable Obj) throws IOException {
		
		File TempFile=new File(this.tempDir,this.tempFileName);
		
		RandomAccessFile DiskCache=new RandomAccessFile(TempFile, "rw");;
		DiskCache.seek(DiskCache.length());

		ByteArrayOutputStream bOP = new ByteArrayOutputStream();
		ObjectWriteStream oO = new ObjectWriteStream(bOP);
		oO.writeObject(Obj);

		byte[] OArray = bOP.toByteArray();
		int ObjectSize=OArray.length;

		DiskCache.write(OArray,0,ObjectSize);
		DiskCache.close();
		
		this.Size++;

	}	

}